package com.wande.dataplatform.service.impl;

import cn.hutool.core.bean.BeanUtil;
import cn.hutool.json.JSONUtil;
import com.baomidou.mybatisplus.core.conditions.query.LambdaQueryWrapper;
import com.baomidou.mybatisplus.core.toolkit.Wrappers;
import com.baomidou.mybatisplus.extension.plugins.pagination.Page;
import com.wande.common.core.exception.ServiceException;
import com.wande.common.core.utils.StringUtils;
import com.wande.common.mybatis.core.page.PageQuery;
import com.wande.common.mybatis.core.page.TableDataInfo;
import com.wande.dataplatform.common.enums.DataPlatformErrorCode;
import com.wande.dataplatform.domain.CleanRule;
import com.wande.dataplatform.domain.CleanTask;
import com.wande.dataplatform.domain.DataModel;
import com.wande.dataplatform.domain.QualityReport;
import com.wande.dataplatform.domain.bo.CleanTaskBo;
import com.wande.dataplatform.domain.dto.CleanResultDTO;
import com.wande.dataplatform.domain.vo.CleanTaskVo;
import com.wande.dataplatform.mapper.CleanRuleMapper;
import com.wande.dataplatform.mapper.CleanTaskMapper;
import com.wande.dataplatform.mapper.DataModelMapper;
import com.wande.dataplatform.mapper.QualityReportMapper;
import com.wande.dataplatform.service.ICleanTaskService;
import com.wande.dataplatform.service.IDorisService;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;

import java.math.BigDecimal;
import java.math.RoundingMode;
import java.time.LocalDateTime;
import java.util.*;

/**
 * 数据清洗任务服务实现
 *
 * @author wande
 */
@Slf4j
@Service
@RequiredArgsConstructor
public class CleanTaskServiceImpl implements ICleanTaskService {

    private final CleanTaskMapper cleanTaskMapper;
    private final CleanRuleMapper cleanRuleMapper;
    private final DataModelMapper dataModelMapper;
    private final QualityReportMapper qualityReportMapper;
    private final IDorisService dorisService;

    @Override
    public TableDataInfo<CleanTaskVo> queryPageList(CleanTaskBo bo, PageQuery pageQuery) {
        LambdaQueryWrapper<CleanTask> wrapper = buildQueryWrapper(bo);
        Page<CleanTask> page = cleanTaskMapper.selectPage(pageQuery.build(), wrapper);
        Page<CleanTaskVo> voPage = new Page<>(page.getCurrent(), page.getSize(), page.getTotal());
        List<CleanTaskVo> voList = BeanUtil.copyToList(page.getRecords(), CleanTaskVo.class);
        // 填充表名
        voList.forEach(vo -> {
            if (vo.getTableId() != null) {
                DataModel model = dataModelMapper.selectById(vo.getTableId());
                if (model != null) {
                    vo.setTableName(model.getTableName());
                }
            }
        });
        voPage.setRecords(voList);
        return TableDataInfo.build(voPage);
    }

    @Override
    public List<CleanTaskVo> queryList(CleanTaskBo bo) {
        LambdaQueryWrapper<CleanTask> wrapper = buildQueryWrapper(bo);
        List<CleanTask> list = cleanTaskMapper.selectList(wrapper);
        return BeanUtil.copyToList(list, CleanTaskVo.class);
    }

    @Override
    public CleanTaskVo queryById(Long id) {
        CleanTask entity = cleanTaskMapper.selectById(id);
        CleanTaskVo vo = BeanUtil.toBean(entity, CleanTaskVo.class);
        if (vo != null && vo.getTableId() != null) {
            DataModel model = dataModelMapper.selectById(vo.getTableId());
            if (model != null) {
                vo.setTableName(model.getTableName());
            }
        }
        return vo;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Long insertByBo(CleanTaskBo bo) {
        CleanTask entity = BeanUtil.toBean(bo, CleanTask.class);
        cleanTaskMapper.insert(entity);
        return entity.getId();
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean updateByBo(CleanTaskBo bo) {
        CleanTask entity = BeanUtil.toBean(bo, CleanTask.class);
        return cleanTaskMapper.updateById(entity) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean deleteByIds(List<Long> ids) {
        return cleanTaskMapper.deleteBatchIds(ids) > 0;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public CleanResultDTO executeClean(Long taskId) {
        long startTime = System.currentTimeMillis();
        CleanTask task = cleanTaskMapper.selectById(taskId);
        if (task == null) {
            throw new ServiceException(DataPlatformErrorCode.ETL_TASK_NOT_FOUND.getMessage());
        }

        // 更新任务状态
        task.setLastExecuteTime(LocalDateTime.now());
        task.setLastExecuteStatus("RUNNING");
        cleanTaskMapper.updateById(task);

        CleanResultDTO result = new CleanResultDTO();
        result.setTaskId(taskId);
        result.setTaskName(task.getName());
        result.setTableId(task.getTableId());

        try {
            // 获取数据模型
            DataModel model = dataModelMapper.selectById(task.getTableId());
            if (model == null) {
                throw new ServiceException("数据模型不存在");
            }
            result.setTableName(model.getTableName());

            // 备份数据
            if (task.getBackupEnabled() != null && task.getBackupEnabled() == 1) {
                String backupTable = task.getBackupTable();
                if (StringUtils.isBlank(backupTable)) {
                    backupTable = model.getTableName() + "_backup_" + System.currentTimeMillis();
                    task.setBackupTable(backupTable);
                }
                backupData(model.getTableName(), backupTable);
                result.setBackupTable(backupTable);
            }

            // 获取规则列表
            List<Long> ruleIds = JSONUtil.toList(task.getRuleIds(), Long.class);
            List<CleanRule> rules = cleanRuleMapper.selectBatchIds(ruleIds);
            // 按优先级排序
            rules.sort(Comparator.comparing(CleanRule::getPriority).reversed());

            // 获取表总行数
            String countSql = String.format("SELECT COUNT(*) as total FROM %s", model.getTableName());
            List<Map<String, Object>> countResult = dorisService.executeQuery(countSql);
            Long totalRows = Long.parseLong(countResult.get(0).get("total").toString());
            result.setTotalRows(totalRows);

            // 执行清洗规则
            List<CleanResultDTO.CleanDetail> cleanDetails = new ArrayList<>();
            Map<String, CleanResultDTO.DimensionStat> dimensionStats = new HashMap<>();
            long totalSuccessRows = 0;
            long totalFailedRows = 0;

            for (CleanRule rule : rules) {
                CleanResultDTO.CleanDetail detail = executeCleanRule(model, rule);
                cleanDetails.add(detail);
                totalSuccessRows += detail.getSuccessRows();
                totalFailedRows += detail.getFailedRows();

                // 统计维度数据
                String dimension = rule.getDimension();
                CleanResultDTO.DimensionStat stat = dimensionStats.computeIfAbsent(dimension, k -> {
                    CleanResultDTO.DimensionStat s = new CleanResultDTO.DimensionStat();
                    s.setDimension(k);
                    s.setRuleCount(0);
                    s.setProcessedRows(0L);
                    return s;
                });
                stat.setRuleCount(stat.getRuleCount() + 1);
                stat.setProcessedRows(stat.getProcessedRows() + detail.getProcessedRows());
            }

            // 计算成功率
            for (CleanResultDTO.DimensionStat stat : dimensionStats.values()) {
                BigDecimal successRate = stat.getProcessedRows() > 0 ?
                    BigDecimal.valueOf(totalSuccessRows)
                        .divide(BigDecimal.valueOf(stat.getProcessedRows()), 4, RoundingMode.HALF_UP)
                        .multiply(BigDecimal.valueOf(100)) : BigDecimal.valueOf(100);
                stat.setSuccessRate(successRate);
            }

            result.setSuccessRows(totalSuccessRows);
            result.setFailedRows(totalFailedRows);
            result.setCleanDetails(cleanDetails);
            result.setDimensionStats(dimensionStats);

            // 计算清洗后质量评分
            BigDecimal afterScore = totalRows > 0 ?
                BigDecimal.valueOf(totalSuccessRows)
                    .divide(BigDecimal.valueOf(totalRows), 4, RoundingMode.HALF_UP)
                    .multiply(BigDecimal.valueOf(100)) : BigDecimal.valueOf(100);
            result.setAfterScore(afterScore);
            result.setBeforeScore(BigDecimal.valueOf(50)); // 简化处理

            result.setStatus("SUCCESS");
            task.setLastExecuteStatus("SUCCESS");

            // 保存质量报告
            saveQualityReport(result, task, model, startTime);

        } catch (Exception e) {
            log.error("执行清洗任务失败", e);
            result.setStatus("FAILED");
            result.setErrorMessage(e.getMessage());
            task.setLastExecuteStatus("FAILED");
        } finally {
            result.setExecuteDuration(System.currentTimeMillis() - startTime);
            cleanTaskMapper.updateById(task);
        }

        return result;
    }

    @Override
    @Transactional(rollbackFor = Exception.class)
    public Boolean rollbackClean(Long taskId) {
        CleanTask task = cleanTaskMapper.selectById(taskId);
        if (task == null || StringUtils.isBlank(task.getBackupTable())) {
            throw new ServiceException("备份表不存在，无法回滚");
        }

        try {
            DataModel model = dataModelMapper.selectById(task.getTableId());
            // 删除当前表数据
            String deleteSql = String.format("TRUNCATE TABLE %s", model.getTableName());
            dorisService.executeUpdate(deleteSql);

            // 从备份表恢复数据
            String restoreSql = String.format("INSERT INTO %s SELECT * FROM %s",
                model.getTableName(), task.getBackupTable());
            dorisService.executeUpdate(restoreSql);

            return true;
        } catch (Exception e) {
            log.error("回滚清洗任务失败", e);
            throw new ServiceException("回滚失败: " + e.getMessage());
        }
    }

    /**
     * 备份数据
     */
    private void backupData(String sourceTable, String backupTable) {
        try {
            // 创建备份表
            String createSql = String.format("CREATE TABLE IF NOT EXISTS %s LIKE %s", backupTable, sourceTable);
            dorisService.executeUpdate(createSql);

            // 复制数据
            String copySql = String.format("INSERT INTO %s SELECT * FROM %s", backupTable, sourceTable);
            dorisService.executeUpdate(copySql);
        } catch (Exception e) {
            log.error("备份数据失败", e);
            throw new ServiceException("备份数据失败: " + e.getMessage());
        }
    }

    /**
     * 执行单个清洗规则
     */
    private CleanResultDTO.CleanDetail executeCleanRule(DataModel model, CleanRule rule) {
        CleanResultDTO.CleanDetail detail = new CleanResultDTO.CleanDetail();
        detail.setRuleId(rule.getId());
        detail.setRuleName(rule.getName());
        detail.setDimension(rule.getDimension());
        detail.setCleanType(rule.getCleanType());

        try {
            // 解析清洗逻辑
            Map<String, Object> cleanLogic = JSONUtil.toBean(rule.getCleanLogic(), Map.class);
            String updateSql = (String) cleanLogic.get("sql");

            if (StringUtils.isNotBlank(updateSql)) {
                // 执行清洗SQL
                int affectedRows = dorisService.executeUpdate(updateSql);
                detail.setProcessedRows((long) affectedRows);
                detail.setSuccessRows((long) affectedRows);
                detail.setFailedRows(0L);
            }

        } catch (Exception e) {
            log.error("执行清洗规则失败: {}", rule.getName(), e);
            detail.setProcessedRows(0L);
            detail.setSuccessRows(0L);
            detail.setFailedRows(0L);
        }

        return detail;
    }

    /**
     * 保存质量报告
     */
    private void saveQualityReport(CleanResultDTO result, CleanTask task, DataModel model, long startTime) {
        QualityReport report = new QualityReport();
        report.setReportType("CLEAN");
        report.setTaskId(task.getId());
        report.setTaskName(task.getName());
        report.setTableId(model.getId());
        report.setTableName(model.getTableName());
        report.setExecuteTime(LocalDateTime.now());
        report.setTotalRows(result.getTotalRows());
        report.setSuccessRows(result.getSuccessRows());
        report.setFailedRows(result.getFailedRows());
        report.setQualityScore(result.getAfterScore());
        report.setExecuteDuration(System.currentTimeMillis() - startTime);
        report.setProblemDetails(JSONUtil.toJsonStr(result.getCleanDetails()));
        report.setDimensionStats(JSONUtil.toJsonStr(result.getDimensionStats()));
        report.setStatus(result.getStatus());
        report.setErrorMessage(result.getErrorMessage());
        report.setProjectId(task.getProjectId());
        qualityReportMapper.insert(report);
    }

    private LambdaQueryWrapper<CleanTask> buildQueryWrapper(CleanTaskBo bo) {
        LambdaQueryWrapper<CleanTask> wrapper = Wrappers.lambdaQuery();
        wrapper.like(StringUtils.isNotBlank(bo.getName()), CleanTask::getName, bo.getName());
        wrapper.eq(bo.getTableId() != null, CleanTask::getTableId, bo.getTableId());
        wrapper.eq(StringUtils.isNotBlank(bo.getScheduleType()), CleanTask::getScheduleType, bo.getScheduleType());
        wrapper.eq(bo.getStatus() != null, CleanTask::getStatus, bo.getStatus());
        wrapper.eq(bo.getProjectId() != null, CleanTask::getProjectId, bo.getProjectId());
        wrapper.orderByDesc(CleanTask::getCreateTime);
        return wrapper;
    }
}
